package com.frank.redis.spring;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.regex.Pattern;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.pool.impl.GenericObjectPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import redis.clients.jedis.JedisShardInfo;
import redis.clients.jedis.ShardedJedis;
import redis.clients.jedis.ShardedJedisPool;
import redis.clients.util.Hashing;

import com.frank.redis.JedisPoolFactory;

/**
 * 支持动态修改shards
 *
 * @author liushaomingdev@163.com
 */
public class JedisPoolBean  extends ShardedJedisPool {

	private static Logger LOG = LoggerFactory.getLogger(JedisPoolBean.class);

	private final GenericObjectPool internalPool;
	public static final String BEAN_PREFIEX = "redis-";
	public static final String DEFAULT_ZKNAME = "default";

	private Hashing algo = Hashing.MURMUR_HASH;
	private Pattern keyTagPattern = Pattern.compile("([0-9]+?)$");

	@Autowired
	private JedisPoolFactory factory;

	private int dbIndex;

	private String zkName = DEFAULT_ZKNAME;

	private volatile List<JedisShardInfo> shards = Collections.emptyList();

	public JedisPoolBean(String conf) {
		super(null, null);
		internalPool = null;
		if (conf.indexOf(':') > 0) {
			String[] values = conf.split(":");
			conf = values[1].trim();

			String name = values[0].trim();
			if (StringUtils.isNotBlank(name))
				zkName = values[0];
		}
		this.dbIndex = Integer.parseInt(conf);
	}

	public void load(Collection<? extends JedisShardInfo> nodes) {
		if (nodes == null)
			return;

		List<JedisShardInfo> shards = new ArrayList<JedisShardInfo>();
		for (JedisShardInfo node : nodes) {
			if (node != null) {
				JedisShardInfo shard = new JedisShardInfo(node.getHost(), node.getPort(), node.getName());
				shard.setPassword(node.getPassword());
				shard.setDb(dbIndex);
				shards.add(shard);
			}
		}
		this.shards = Collections.unmodifiableList(shards);
	}

	@Override
	public ShardedJedis getResource() {
		if (internalPool == null)
			createPool();

		Jedis jedis = null;
		while (true) {
			jedis = (Jedis) super.getResource();
			if (jedis == null || jedis.shards == shards)
				break;
			super.returnBrokenResourceObject(jedis);
		}
		return jedis;
	}

	@Override
	public void returnResourceObject(Object resource) {
		if (resource == null)
			return;

		Jedis jedis = (Jedis) resource;
		if (jedis.shards == shards)
			super.returnResourceObject(resource);
		else
			super.returnBrokenResourceObject(resource);
	}

	private synchronized void createPool() {
		if (internalPool == null) {
			internalPool = factory.createPool(new JedisFactory());
			LOG.debug("load jedis({}) pool: {}", dbIndex, shards);
		}
	}

	public String getZkName() {
		return zkName;
	}

	public int getSize() {
		return internalPool.getNumActive() + internalPool.getNumIdle();
	}

	private class JedisFactory extends ShardedJedisFactory {
		public JedisFactory() {
			super(null, null, null);
		}

		@Override
		public Object makeObject() throws Exception {
			LOG.debug("create jedis({}) {}'s connection", dbIndex, getSize()+1);

			JedisPoolBean pool = JedisPoolBean.this;
			return new Jedis(pool.shards, pool.algo, pool.keyTagPattern);
		}

		@Override
		public void destroyObject(Object obj) throws Exception {
			LOG.debug("destroy jedis({}) connection", dbIndex);
			super.destroyObject(obj);
		}
	}

	private class Jedis extends redis.clients.jedis.ShardedJedis {

		List<JedisShardInfo> shards;

		public Jedis(List<JedisShardInfo> shards, Hashing algo,
				Pattern keyTagPattern) {
			super(shards, algo, keyTagPattern);
			this.shards = shards;
		}
	}
}
